package com.example.rocketmqc.Bean;

import com.alibaba.fastjson.JSONException;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @author CHENTAIWEN
 * @create 2022/8/9 16:11
 */
@Slf4j
@Configuration
public class SpringConfig {

    @Value("${rocketmq.consumer.groupName}")
    private String groupName;
    @Value("${rocketmq.consumer.namesrvAddr}")
    private String namesrvAddr;
    @Value("${rocketmq.consumer.instanceName}")
    private String instanceName;
    @Value("${rocketmq.topic}")
    private String topic;
    @Value("${rocketmq.tag}")
    private String tag;

    @Bean
    public DefaultMQPushConsumer getRocketMQConsumer() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(this.groupName);
        consumer.setNamesrvAddr(this.namesrvAddr);
        try {
            consumer.subscribe(this.topic, tag);
            consumer.registerMessageListener((MessageListenerOrderly) (msgList, context) -> {
                try {
                    Message msg = null;
                    for (Message aMsgList : msgList) {
                        msg = aMsgList;
                        try {
                            Thread.sleep(2000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        System.out.println("收到MQ消息："+msg);
                        // rocketMQ获取的消息类型是byte[] 如果使用该方法需要对消息类型进行转化
                        String s = new String(msg.getBody(), StandardCharsets.UTF_8);
                        log.error("消息内容:{}",s);
                    }

                } catch (JSONException e) {
                    System.out.println(e);
                }

                return ConsumeOrderlyStatus.SUCCESS;
            });
            consumer.start();
            System.out.println("已启动Conusmer【gruop:" + this.groupName + "，instance:" + this.instanceName
                    + "】，监听TOPIC-{" + this.topic + "},tag-{" + this.tag + "}");
        } catch (MQClientException e) {
            System.out.println(e);
        }
        return consumer;
    }
}
